package com.ada.flink

import com.ada.flink.bean.StartUpLog
import com.ada.flink.util.{MyEsUtil, MyJdbcSink, MyKafkaUtil, MyRedisUtil}
import com.alibaba.fastjson.JSON
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, KeyedStream, SplitStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011
import org.apache.flink.streaming.connectors.redis.RedisSink

object StreamApiApp {

    def main(args: Array[String]): Unit = {
        val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

        val kafkaConsumer = MyKafkaUtil.getConsumer("GMALL_STARTUP")

        import org.apache.flink.api.scala._

        val dstream: DataStream[String] = environment.addSource(kafkaConsumer)

        //dstream.print()
        //求各个渠道的累计个数
        val startUplogDstream: DataStream[StartUpLog] = dstream.map {
            JSON.parseObject(_, classOf[StartUpLog])
        }
        /*
        keyBy:
        DataStream→KeyedStream：输入必须是Tuple类型，逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同key的元素，在内部以hash的形式实现的。
         */
        val keyedStream: KeyedStream[(String, Int), Tuple] = startUplogDstream.map(startuplog => (startuplog.ch, 1)).keyBy(0)

        val sumDstream: DataStream[(String, Int)] = keyedStream.sum(1)
        /*
        reduce:
        KeyedStream→ DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
         */
        //reduce //sum
        keyedStream.reduce { (ch1, ch2) =>
            (ch1._1, ch1._2 + ch2._2)
        }.print().setParallelism(1)
        /*
        flink是如何保存累计值的，
        flink是一种有状态的流计算框架，其中说的状态包括两个层面：
        1)  operator state 主要是保存数据在流程中的处理状态，用于确保语义的exactly-once。
        2)  keyed state  主要是保存数据在计算过程中的累计值。
        这两种状态都是通过checkpoint机制保存在StateBackend中，StateBackend可以选择保存在内存中（默认使用）或者保存在磁盘文件中。
         */

        /*
        split:
        DataStream →SplitStream：根据某些特征把一个DataStream拆分成两个或者多个DataStream。
         */
        // 将appstore与其他渠道拆分拆分出来  成为两个独立的流
        val splitStream: SplitStream[StartUpLog] = startUplogDstream.split { startUplog =>
            var flags: List[String] = null
            if ("appstore" == startUplog.ch) {
                flags = List(startUplog.ch)
            } else {
                flags = List("other")
            }
            flags
        }
        /*
        Select:
        SplitStream→DataStream：从一个SplitStream中获取一个或者多个DataStream。
         */
        val appStoreStream: DataStream[StartUpLog] = splitStream.select("appstore")
        appStoreStream.print("apple:").setParallelism(1)
        val otherStream: DataStream[StartUpLog] = splitStream.select("other")
        otherStream.print("other:").setParallelism(1)

        /*
        Connect和CoMap:
        DataStream,DataStream→ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被Connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。
         */
        val connStream: ConnectedStreams[StartUpLog, StartUpLog] = appStoreStream.connect(otherStream)
        val allStream: DataStream[String] = connStream.map(
            (log1: StartUpLog) => log1.ch,
            (log2: StartUpLog) => log2.ch
        )
        allStream.print("connect::")

        /*
        union:
        DataStream → DataStream：对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream。注意:如果你将一个DataStream跟它自己做union操作，在新的DataStream中，你将看到每一个元素都出现两次。
         */
        val unionStream: DataStream[StartUpLog] = appStoreStream.union(otherStream)
        unionStream.print("union:::")
        /*
        Connect与 Union 区别：
        1、Union之前两个流的类型必须是一样，Connect可以不一样，在之后的coMap中再去调整成为一样的。
        2、Connect只能操作两个流，Union可以操作多个
         */

        //kafka sink
        val kafkaSink: FlinkKafkaProducer011[String] = MyKafkaUtil.getProducer("appstore")
        sumDstream.map(chCount => chCount._1 + ":" + chCount._2).addSink(kafkaSink)

        //redis sink
        val redisSink: RedisSink[(String, String)] = MyRedisUtil.getRedisSink()
        keyedStream.map(chCount => (chCount._1, chCount._2.toString)).addSink(redisSink)

        //Elasticsearch sink
        val esSink: ElasticsearchSink[String] = MyEsUtil.getElasticSearchSink("gmall_startup")
        dstream.addSink(esSink)

        //自定义JDBC sink
        val jdbcSink = new MyJdbcSink("insert into appstore_startup values(?,?,?,?,?)")
        appStoreStream.map(startuplog => Array(startuplog.mid, startuplog.uid, startuplog.ch, startuplog.area, startuplog.ts)).addSink(jdbcSink)

        environment.execute()
    }

}
